package cn.com.chengmandian.core.lot;

import cn.com.chengmandian.core.lot.utils.AmqpUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import javax.jms.*;
import javax.naming.Context;

/**
 * AMQP订阅消息
 */
@Slf4j
@Component
public class AmqpService {

    @Autowired(required = false)
    @Qualifier("amqpJmsConnectionListener")
    private JmsConnectionListener jmsConnectionListener;

    @Autowired(required = false)
    @Qualifier("amqpMessageListener")
    private MessageListener messageListener;

    /**
     * 创建消费连接
     */
    public void createConnections(String accessKeyId, String accessKeySecret, String host, String iotInstanceId, Integer maxConnection, String consumerGroupId) {
        int success = 0;
        for (int i = 0; i <maxConnection; i ++) {
            try {
                this.createSingleConnection(accessKeyId, accessKeySecret, host, iotInstanceId, consumerGroupId);
                success = success + 1;
            } catch (Exception e) {
                log.error("AmqpUtil start failed：{}", consumerGroupId);
            }
        }
        log.info("AmqpUtil：{}", "成功启动" + success + "个连接");
    }

    /**
     * 创建连接，可创建多个
     */
    public void createSingleConnection(String accessKeyId, String accessKeySecret, String host, String iotInstanceId, String consumerGroupId) throws Exception {
        if (jmsConnectionListener == null) {
            throw new RuntimeException("amqp need jmsConnectionListener, please implements org.apache.qpid.jms.JmsConnectionListener!");
        }
        if (messageListener == null) {
            throw new RuntimeException("amqp need amqpMessageListener, please implements javax.jms.MessageListener!");
        }
        Context context = AmqpUtil.initContext(host);
        Destination queue = (Destination) context.lookup("QUEUE");

        JmsConnection connection = AmqpUtil.initJmsConnection(context, iotInstanceId, accessKeyId, accessKeySecret, consumerGroupId);
        // 添加连接监听器
        connection.addConnectionListener(jmsConnectionListener);
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 自动ACK
        connection.start();
        // 创建Receiver连接。
        MessageConsumer consumer = session.createConsumer(queue);
        consumer.setMessageListener(messageListener);
    }

}
